package com.loading;


import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.RestOptions;
import org.apache.flink.configuration.TaskManagerOptions;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

import java.io.*;
import java.net.ServerSocket;
import java.net.Socket;


public class BaseServer {

    public static void base(String[] args) throws IOException {
        ServerSocket serverSocket = new ServerSocket(Integer.parseInt(args[3]));
        System.out.println("Server is running on port " + Integer.parseInt(args[3]) + "...");

        while(true) {
            Socket socket = serverSocket.accept();
            System.out.println("New client connected: " + socket.getInetAddress());
            OutputStream outputStream = socket.getOutputStream();
            PrintWriter out = new PrintWriter(outputStream, true);
            InputStream inputStream = socket.getInputStream();
            BufferedReader in = new BufferedReader(new InputStreamReader(inputStream));

            String inputLine;
            while((inputLine = in.readLine()) != null) {
                System.out.println("Received message from client: " + inputLine);
                out.println("Echo from server: " + inputLine);
            }

            socket.close();
        }
    }

    public static void main(String[] args) throws Exception {
        (new Thread(() -> {
            try {
                base(args);
            } catch (IOException var2) {
                var2.printStackTrace();
            }

        })).start();
        Configuration config = new Configuration();
        config.set(RestOptions.BIND_PORT, args[0]);
        config.setInteger(TaskManagerOptions.NUM_TASK_SLOTS, Integer.parseInt(args[1]));
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(config);
        DataStreamSource<String> socketDS = env.socketTextStream(args[2], Integer.parseInt(args[3]));
        // TODO 3. 处理数据: 切换、转换、分组、聚合
        SingleOutputStreamOperator<Tuple2<String, Integer>> sum = socketDS
                .flatMap(
                        (String value, Collector<Tuple2<String, Integer>> out) -> {
                            String[] words = value.split(" ");
                            for (String word : words) {
                                out.collect(Tuple2.of(word, 1));
                            }
                        }
                )
                .setParallelism(2)
                .returns(Types.TUPLE(Types.STRING,Types.INT))
//                .returns(new TypeHint<Tuple2<String, Integer>>() {})
                .keyBy(value -> value.f0)
                .sum(1);



        // TODO 4. 输出
        sum.print();

        // TODO 5. 执行
        env.execute("system");
    }
}

/**

 并行度的优先级：
    代码：算子 > 代码：env > 提交时指定 > 配置文件

 */
